package com.tdy.cdc.app.cdr;

import com.tdy.cdc.app.BaseSqlApp;
import com.tdy.cdc.common.Constant;
import com.tdy.cdc.util.FlinkSinkUtil;
import com.tdy.cdc.util.FlinkSourceUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: NanHuang
 * @Date: 2023/04/10/16:43
 * @Description: DWD层：dwd_mhs_exam_report 表手动分流
 */
public class Cdr_Hdsd00_05_02 extends BaseSqlApp {
    public static void main(String[] args) {
        new Cdr_Hdsd00_05_02().init(
                6005,
                "Cdr_Hdsd00_05_02",
                2
        );
    }
    @Override
    protected void invoke(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        super.readVisit(tableEnv);
        super.readPatient(tableEnv);

        tableEnv.executeSql("create table dwd_mhs_exam_report (" +
                "XDS_TYPE                       STRING," +
                "XDS_NAME                       STRING," +
                "XDS_VERSION                    STRING," +
                "PATIENT_DOMAIN                 STRING," +
                "VISIT_DOMAIN                   STRING," +
                "CUSTODIAN_CODE                 STRING," +
                "CUSTODIAN_NAME                 STRING," +
                "HOSPITAL_CODE                  STRING," +
                "HOSPITAL_NAME                  STRING," +
                "INPATIENT_VISIT_FLOW_ID        STRING," +
                "OUTPATIENT_VISIT_FLOW_ID       STRING," +
                "VISIT_TIMES                    STRING," +
                "PATIENT_ID                     STRING," +
                "RECORD_ID                      STRING," +
                "DW_DATE                        STRING," +
                "DOMAIN_ID                      STRING," +
                "OUTPATIENT_NO                  STRING," +
                "INPATIENT_NO                   STRING," +
                "PATIENT_VISIT_CATEGORY_CODE    STRING," +
                "PATIENT_VISIT_CATEGORY         STRING," +
                "EXAM_REPORT_NO                 STRING," +
                "APPLY_SHEET_NO                 STRING," +
                "SPECIMEN_NUMBER                STRING," +
                "TELECOM                        STRING," +
                "ID_CARD_NUMBER                 STRING," +
                "PATIENT_NAME                   STRING," +
                "SEX_NAME                       STRING," +
                "SEX_CODE                       STRING," +
                "AGE                            STRING," +
                "AGE_UNIT                       STRING," +
                "EXAM_REPORT_TIME               STRING," +
                "AUTHOR_CODE                    STRING," +
                "AUTHOR_NAME                    STRING," +
                "CUSTODIAN_ADDR                 STRING," +
                "AUDIT_SIGNDATE                 STRING," +
                "EXAM_AUDIT_DOCTOR_ID           STRING," +
                "EXAM_AUDIT_DOCTOR_NAME         STRING," +
                "EXAM_AUDIT_DATE                STRING," +
                "EXAM_TECH_STAFF_ID             STRING," +
                "EXAM_TECH_STAFF_NAME           STRING," +
                "TECHNICIAN_SIGNDATE            STRING," +
                "EXAM_DOCTOR_ID                 STRING," +
                "EXAM_DOCTOR_NAME               STRING," +
                "APPLY_DATE                     STRING," +
                "APPLY_DEPT_ID                  STRING," +
                "APPLY_DEPT                     STRING," +
                "APPLY_AGENCYCODE               STRING," +
                "APPLY_ORGAN                    STRING," +
                "VISIT_DATE                     STRING," +
                "BED_NO                         STRING," +
                "ROOM_NO                        STRING," +
                "ROOM_NAME                      STRING," +
                "WARDS_CODE                     STRING," +
                "BED_NAME                       STRING," +
                "DEPT_ID                        STRING," +
                "DEPT_NAME                      STRING," +
                "WARDS_NAME                     STRING," +
                "CHIEF_COMPLAINT                STRING," +
                "SYMPTOM_START_TIME             STRING," +
                "SYMPTOM_END_TIME               STRING," +
                "SYMPTOM_DESC                   STRING," +
                "EXAM_SPECIAL_FLAG              STRING," +
                "CHECK_METHOD                   STRING," +
                "INSPECT_TYPE                   STRING," +
                "TREAT_PROCESS_DESC             STRING," +
                "EXAM_REPORT_RESULT_FINDINGS    STRING," +
                "EXAM_REPORT_RESULT_HINT        STRING," +
                "EXAM_REPORT_DEPT_NAME          STRING," +
                "AGENCY_NAME                    STRING," +
                "REPORT_REMARK                  STRING," +
                "INSPECT_TYPE_CODE              STRING," +
                "EXAM_DATE                      STRING," +
                "ITEM_CODE                      STRING," +
                "ITEM_NAME                      STRING," +
                "SPECIMEN_SAMPLE_TIME           STRING," +
                "SPECIMEN_RECV_TIME             STRING," +
                "SPECIMEN_CLASS                 STRING," +
                "SPECIMEN_STATUS                STRING," +
                "SPECIMEN_FIXATIVE_NAME         STRING," +
                "RESULT_NAME                    STRING," +
                "RESULT_CODE                    STRING," +
                "EXAM_RESULT_VALUE              STRING," +
                "EXAM_RESULT_UNIT               STRING," +
                "DIAG_DATE                      STRING," +
                "DIAG_CODE                      STRING," +
                "DIAG_NAME                      STRING," +
                "DIAG_TYPE                      STRING," +
                "DIAG_INSTITUTION               STRING," +
                "`pt` as proctime()" +
                ")" +
                FlinkSourceUtil.getKafkaWith(Constant.TOPIC_DWD_MHS_EXAM_REPORT,"test"));

        // cdr_hdsd00_05_02
        tableEnv.executeSql("create table cdr_hdsd00_05_02(" +
                "xds_type               string," +
                "xds_name               string," +
                "xds_version            string," +
                "patient_domain         string," +
                "visit_domain           string," +
                "custodian_code         string," +
                "custodian_name         string," +
                "hospital_code          string," +
                "hospital_name          string," +
                "visit_id               string," +
                "visit_times            string," +
                "patient_id             string," +
                "unique_id              string," +
                "effective_time         string," +
                "domain_id              string," +
                "out_patient_id         string," +
                "inpatient              string," +
                "patient_typecode       string," +
                "patient_type           string," +
                "report_id              string," +
                "appy_id                string," +
                "specimen_number        string," +
                "telecom                string," +
                "identify_no            string," +
                "patient_name           string," +
                "sex                    string," +
                "sex_code               string," +
                "age                    string," +
                "age_unit               string," +
                "write_time             string," +
                "author_code            string," +
                "author_name            string," +
                "custodian_addr         string," +
                "audit_signdate         string," +
                "audit_code             string," +
                "audit_name             string," +
                "check_signdate         string," +
                "technician_code        string," +
                "technician_name        string," +
                "technician_signdate    string," +
                "check_code             string," +
                "check_name             string," +
                "apply_date             string," +
                "apply_dept_code        string," +
                "apply_dept             string," +
                "apply_agencycode       string," +
                "apply_organ            string," +
                "visit_date             string," +
                "bed_no                 string," +
                "ward_id                string," +
                "ward_name              string," +
                "wards_id               string," +
                "bed_name               string," +
                "dept_code              string," +
                "dept_name              string," +
                "wards_name             string," +
                "chief_complaint        string," +
                "symptom_start_date     string," +
                "symptom_stopping_date  string," +
                "symptom_description    string," +
                "special_exam_mark      string," +
                "check_method           string," +
                "inspect_type           string," +
                "treatment_process      string," +
                "objective_observation  string," +
                "subjective_tip         string," +
                "report_dept            string," +
                "agency_name            string," +
                "report_remark          string," +
                "inspect_type_code      string," +
                "status                 string," +
                "empi                   string," +
                "encounter_id           string " +
                ")" +
                FlinkSinkUtil.getKafkaWith("cdr_hdsd00_05_02"));

        // cdr_tdy_list_diag
        tableEnv.executeSql("create table cdr_tdy_list_diag(" +
                "unique_id              string," +
                "xds_type               string," +
                "xds_version            string," +
                "domain_id              string," +
                "diag_code              string," +
                "diag_name              string," +
                "diag_type              string," +
                "diag_institution       string " +
                ")" +
                FlinkSinkUtil.getKafkaWith("cdr_tdy_list_diag"));

        // cdr_tdy_list_exam
        tableEnv.executeSql("create table cdr_tdy_list_exam(" +
                "unique_id              string," +
                "xds_type               string," +
                "xds_version            string," +
                "domain_id              string," +
                "check_date             string," +
                "item_code              string," +
                "item_name              string," +
                "sample_date            string," +
                "receipt_date           string," +
                "specimen_category      string," +
                "specimen_status        string," +
                "specimen_fixative      string," +
                "result_name            string," +
                "result_code            string," +
                "quanty_result          string," +
                "quanty_unit            string " +
                ")" +
                FlinkSinkUtil.getKafkaWith("cdr_tdy_list_exam"));

        tableEnv.createStatementSet().
                addInsertSql("insert into cdr_hdsd00_05_02 " +
                        "select " +
                        "   'C0006'                      ," +
                        "   '检查报告'                    ," +
                        "   'v1.0.0.0'                   ," +
                        "   t.PATIENT_DOMAIN             ," +
                        "   t.VISIT_DOMAIN               ," +
                        "   t.CUSTODIAN_CODE             ," +
                        "   t.CUSTODIAN_NAME             ," +
                        "   t.HOSPITAL_CODE              ," +
                        "   t.HOSPITAL_NAME              ," +
                        "   case when t.PATIENT_VISIT_CATEGORY_CODE='3' then t.INPATIENT_VISIT_FLOW_ID else t.OUTPATIENT_VISIT_FLOW_ID end ," +
                        "   t.VISIT_TIMES                ," +
                        "   t.PATIENT_ID                 ," +
                        "   t.EXAM_REPORT_NO                  ," +
                        "   t.DW_DATE                    ," +
                        "   t.DOMAIN_ID                  ," +
                        "   t.OUTPATIENT_NO              ," +
                        "   t.INPATIENT_NO               ," +
                        "   t.PATIENT_VISIT_CATEGORY_CODE," +
                        "   t.PATIENT_VISIT_CATEGORY     ," +
                        "   t.EXAM_REPORT_NO             ," +
                        "   t.APPLY_SHEET_NO             ," +
                        "   t.SPECIMEN_NUMBER            ," +
                        "   t.TELECOM                    ," +
                        "   t.ID_CARD_NUMBER             ," +
                        "   t.PATIENT_NAME               ," +
                        "   t.SEX_NAME                   ," +
                        "   t.SEX_CODE                   ," +
                        "   t.AGE                        ," +
                        "   t.AGE_UNIT                   ," +
                        "   t.EXAM_REPORT_TIME           ," +
                        "   t.AUTHOR_CODE                ," +
                        "   t.AUTHOR_NAME                ," +
                        "   t.CUSTODIAN_ADDR             ," +
                        "   t.AUDIT_SIGNDATE             ," +
                        "   t.EXAM_AUDIT_DOCTOR_ID       ," +
                        "   t.EXAM_AUDIT_DOCTOR_NAME     ," +
                        "   t.EXAM_AUDIT_DATE            ," +
                        "   t.EXAM_TECH_STAFF_ID         ," +
                        "   t.EXAM_TECH_STAFF_NAME       ," +
                        "   t.TECHNICIAN_SIGNDATE        ," +
                        "   t.EXAM_DOCTOR_ID             ," +
                        "   t.EXAM_DOCTOR_NAME           ," +
                        "   t.APPLY_DATE                 ," +
                        "   t.APPLY_DEPT_ID              ," +
                        "   t.APPLY_DEPT                 ," +
                        "   t.APPLY_AGENCYCODE           ," +
                        "   t.APPLY_ORGAN                ," +
                        "   t.VISIT_DATE                 ," +
                        "   t.BED_NO                     ," +
                        "   t.ROOM_NO                    ," +
                        "   t.ROOM_NAME                  ," +
                        "   t.WARDS_CODE                 ," +
                        "   t.BED_NAME                   ," +
                        "   t.DEPT_ID                    ," +
                        "   t.DEPT_NAME                  ," +
                        "   t.WARDS_NAME                 ," +
                        "   t.CHIEF_COMPLAINT            ," +
                        "   t.SYMPTOM_START_TIME         ," +
                        "   t.SYMPTOM_END_TIME           ," +
                        "   t.SYMPTOM_DESC               ," +
                        "   t.EXAM_SPECIAL_FLAG          ," +
                        "   t.CHECK_METHOD               ," +
                        "   t.INSPECT_TYPE               ," +
                        "   t.TREAT_PROCESS_DESC         ," +
                        "   t.EXAM_REPORT_RESULT_FINDINGS," +
                        "   t.EXAM_REPORT_RESULT_HINT    ," +
                        "   t.EXAM_REPORT_DEPT_NAME      ," +
                        "   t.AGENCY_NAME                ," +
                        "   t.REPORT_REMARK              ," +
                        "   t.INSPECT_TYPE_CODE          ," +
                        "   if(p.PATIENT_ID is not null and (v1.VISIT_ID is not null or v2.VISIT_ID is not null),'AA','AE')            ," +
                        "   p.GLOBAL_ID                                                                                                ," +
                        "   case " +
                        "      when v1.VISIT_ID is not null then v1.VISIT_DOMAIN_ID || '_' || v1.VISIT_ID || '_' || v1.PATIENT_CATEGORY " +
                        "      when v2.VISIT_ID is not null then v2.VISIT_DOMAIN_ID || '_' || v2.VISIT_ID || '_' || v2.PATIENT_CATEGORY " +
                        "   end  " +
                        "from dwd_mhs_exam_report t                                                                         " +
                        "left join patient for system_time as of t.pt as p on t.PATIENT_ID = p.PATIENT_ID                   " +
                        "left join visit for system_time as of t.pt as v1 on t.INPATIENT_VISIT_FLOW_ID = v1.VISIT_ID        " +
                        "left join visit for system_time as of t.pt as v2 on t.OUTPATIENT_VISIT_FLOW_ID = v2.VISIT_ID       ")
                .addInsertSql("insert into cdr_tdy_list_diag " +
                        "select " +
                        "   EXAM_REPORT_NO        ," +
                        "   'C0006'               ," +
                        "   'v1.0.0.0'            ," +
                        "   DOMAIN_ID             ," +
                        "   DIAG_CODE             ," +
                        "   DIAG_NAME             ," +
                        "   DIAG_TYPE             ," +
                        "   DIAG_INSTITUTION       " +
                        "from dwd_mhs_exam_report  " +
                        "where DIAG_CODE is not null and DIAG_CODE <> '' ")
                .addInsertSql("insert into cdr_tdy_list_exam " +
                        "select " +
                        "   EXAM_REPORT_NO        ," +
                        "   'C0006'               ," +
                        "   'v1.0.0.0'            ," +
                        "   DOMAIN_ID             ," +
                        "   EXAM_DATE             ," +
                        "   ITEM_CODE             ," +
                        "   ITEM_NAME             ," +
                        "   SPECIMEN_SAMPLE_TIME  ," +
                        "   SPECIMEN_RECV_TIME    ," +
                        "   SPECIMEN_CLASS        ," +
                        "   SPECIMEN_STATUS       ," +
                        "   SPECIMEN_FIXATIVE_NAME," +
                        "   RESULT_NAME           ," +
                        "   RESULT_CODE           ," +
                        "   EXAM_RESULT_VALUE     ," +
                        "   EXAM_RESULT_UNIT       " +
                        "from dwd_mhs_exam_report  " +
                        "where ITEM_CODE is not null and ITEM_CODE <> '' ")
                .execute();
    }
}
